package own.stu.jobgib.playown.redis.redisson;

import lombok.extern.slf4j.Slf4j;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;

@Slf4j
public class LockTest {

    public static void main(String[] args) throws Exception {
//        testReentrantLock();
//        testRLock();

//        rwLock(true);
        rwLock(false);

    }

    private static void rwLock(boolean writeFirst) throws Exception {
        RedissonClient client = getRedissonClient();
        final RReadWriteLock readWriteLock = client.getReadWriteLock("rw-lock");

        final ExecutorService service = Executors.newFixedThreadPool(5);
        long defaultSleep = 200;
        AtomicLong writeSleep = new AtomicLong(defaultSleep);
        if (writeFirst) {
            writeSleep.compareAndSet(defaultSleep, defaultSleep - 50);
        } else {
            writeSleep.compareAndSet(defaultSleep, defaultSleep + 50);
        }
        service.submit(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(writeSleep.get());
                log.info("{} write {}", Thread.currentThread().getName(), writeSleep.get());
                readWriteLock.writeLock().lock();
                TimeUnit.MINUTES.sleep(5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                readWriteLock.writeLock().unlock();
            }
        });

        for (int i = 0; i < 2; i++)
            service.submit(() -> {

                try {
                    TimeUnit.MILLISECONDS.sleep(defaultSleep);
                    log.info("{} read {}", Thread.currentThread().getName(), defaultSleep);
                    readWriteLock.readLock().lock();
                    TimeUnit.MINUTES.sleep(3);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    readWriteLock.readLock().unlock();
                }
            });

        TimeUnit.MINUTES.sleep(10);

//        service.shutdown();
//        client.shutdown();
    }

    private static void testReentrantLock() throws Exception {
        ReentrantLock fairLock = new ReentrantLock(true);
        testFairLock(fairLock);
    }

    private static void testRLock() throws Exception {
        final RLock fairLock = getLock();
        testFairLock(fairLock);
    }

    private static RedissonClient getRedissonClient() {
        Config config = new Config();
        config.useSingleServer()
                /*

                redis-server conf配置文件 ： maxclients 20 (客户端最大的连接数数量）
                客户端超出，会出现以下异常：
                Unable to init enough connections amount! Only 19 of 24 were initialized. Redis server: localhost/127.0.0.1:6379

                查看客户端的3306端口 打开文件数，即打开的连接
                lsof -i:6379 | grep java
                 */
                .setConnectionMinimumIdleSize(4)
                // use "rediss://" for SSL connection
                //.setAddress("redis://rediss.server:6379"); // 修改hosts文件
                .setAddress("redis://localhost:6379"); // 修改hosts文件
        // Sync and Async API
        return Redisson.create(config);
    }
    private static RLock getLock() {
        RedissonClient redisson = getRedissonClient();

        // mock 公平锁的唤醒问题，第二个线程挂了，后续的线程还会继续获取锁吗， 会 太慢了，还没找到超时时间的设置 TODO
        // fair lock
        return redisson.getFairLock("f-lock");
    }

    private static void testFairLock(Lock fairLock) throws InterruptedException {

        new Thread(run(1, fairLock, (v) -> {
            log.info("[{}] do business", Thread.currentThread().getName());
            try {
                TimeUnit.SECONDS.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }), "t-1").start();

        Thread t2 = new Thread(run(2, fairLock, (v) -> {
            log.info("[{}] do business", Thread.currentThread().getName());
        }), "t-2");
        t2.start();

        new Thread(run(3, fairLock, (v) -> {
            log.info("[{}] do business", Thread.currentThread().getName());
        }), "t-3").start();
        TimeUnit.MILLISECONDS.sleep(240);
        t2.interrupt();

        System.out.println("main wait.....");
        TimeUnit.SECONDS.sleep(20);
    }

    private static Runnable run(int seq, Lock fairLock, Consumer<Void> consumer) {
        return () -> {
            try {
                log.info("[{}] try getLock ", Thread.currentThread().getName());
                TimeUnit.MILLISECONDS.sleep(100 * seq);
                fairLock.lockInterruptibly();
                log.info("[{}] getLock ", Thread.currentThread().getName());
                consumer.accept(null);
            } catch (Exception e) {
                log.error("lock error: ", e);
                Thread.currentThread().interrupt();
            } finally {
                try {
                    fairLock.unlock();
                } catch (Exception e) {
                    log.info("[{}] unLock error ", Thread.currentThread().getName(), e);
                }
                log.info("[{}] out ", Thread.currentThread().getName());
            }
        };
    }

}
